package com.autodesk.akn.emr.util;

import com.autodesk.akn.emr.dao.S3DaoImpl;
import com.autodesk.akn.emr.util.s3nativefilesystem.Jets3tNativeFileSystemStore;
import com.autodesk.akn.emr.util.s3nativefilesystem.NativeFileSystemStore;
import com.autodesk.akn.emr.util.s3nativefilesystem.NativeS3FileSystem;
import com.autodesk.akn.emr.web.entity.SPSSConversionParams;
import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.log4j.Logger;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class S3Util {
    private final static Logger LOGGER = Logger.getLogger(S3Util.class.getName());
    public static final String SAV_EXT = ".sav";

    private NativeS3FileSystem fs;
    private SPSSUtil spssUtil = new SPSSUtil();
    public void setFileSystem(String inputBucket) throws IOException {
        //Obtain the access key and the secret key from the property file
        Properties prop = new Properties();
        prop.load(S3Util.class.getClassLoader().getResourceAsStream("config.properties"));
        String accessKey = (String) prop.get("credentials.accessKey");
        String secretKey = (String) prop.get("credentials.secretKey");
        String proxy =(String)prop.get("proxy.enabled");
        boolean proxyneeded = "true".equals(proxy);
        //Add the keys to the conf object and pass it to the file system
        Configuration conf = new Configuration();
        conf.set("fs.s3n.awsAccessKeyId", accessKey);
        conf.set("fs.s3n.awsSecretAccessKey", secretKey);
        URI uri = URI.create("s3n://"+inputBucket+"/");
        HostConfiguration hostConfiguration = new HostConfiguration();
        if(proxyneeded){
            hostConfiguration.setProxy("proxy.corp.globant.com",3128);
        }
        NativeFileSystemStore store = new Jets3tNativeFileSystemStore();
        fs = new NativeS3FileSystem(store);
        fs.initialize(uri,conf,hostConfiguration);
        LOGGER.info("File system detected and initialized");
    }

    public List<String> copyFromS3(boolean isFolder,
                                   SPSSConversionParams params,
                                   List<String> filesToProcess) throws IOException{
        List<String>localFiles = new ArrayList<String>();
        if(isFolder){
            for(String file: filesToProcess){
                if(file.contains(SAV_EXT)){
                   LOGGER.info("File Name: "+file);
                   localFiles.add(copySingleFileFromS3(params.getInputBucket(),params.getInputKey()+file));
                }
            }
        } else {
            localFiles.add(copySingleFileFromS3(params.getInputBucket(),params.getInputKey()));
        }
        return localFiles;
    }

    public String copySingleFileFromS3(String inputBucket, String inputKey) {
        String[]splittedName = inputKey.split("/");
        String fileName = splittedName[splittedName.length-1];
        FSDataInputStream s3InputStream = null;
        FileOutputStream localFileOutputStream = null;
        try{
            Path inFile = new Path("s3n://"+inputBucket+"/"+inputKey);
            s3InputStream = fs.open(inFile);
            File inputFile = File.createTempFile(fileName,"");
            localFileOutputStream = new FileOutputStream(inputFile);

            IOUtils.copy(s3InputStream, localFileOutputStream);
            fileName = inputFile.getPath();
            LOGGER.info("Copied file: "+fileName);
        }catch(IOException e){
            LOGGER.error(e.getMessage());
        }finally {
            try{
                if(s3InputStream!=null) s3InputStream.close();
                if(localFileOutputStream!=null) localFileOutputStream.close();
            } catch (IOException e){
                LOGGER.error(e.getMessage());
            }
        }
        return fileName;
    }

    public List<String> convertSAV2CSV(List<String> files, boolean headersOnly, String outputExtension){
        List<String>exportedFiles=new ArrayList<String>();
        LOGGER.info("Begin Exportation");
        for(String file : files){
            String outputFileName = null;
            try{
                //Create a local output file to be used by converter
                String inputFile = file;
                file = file.substring(0,file.indexOf(SAV_EXT));
                outputFileName = file + outputExtension;
                File outputFile = new File(outputFileName);
                //Perform the exportation
                spssUtil.export(new File(inputFile), outputFile, headersOnly);
                exportedFiles.add(outputFileName);
            }catch (Exception e){
                LOGGER.error("Error trying to export "+outputFileName+" "+e.getMessage());
            }
        }
        LOGGER.info("End Exportation");
        return exportedFiles;
    }

    public void copyToS3(List<String> exportedFiles,
                         String outputBucket,
                         String outputKey,
                         boolean isFolder){
        LOGGER.info("begin copy from local to S3");
        for(String file:exportedFiles){
            FSDataOutputStream s3OutputStream = null;
            FileInputStream localFileInputStream = null;
            try{
                Path outFile = null;
                if(isFolder){
                    outFile = new Path("s3n://"+outputBucket+"/"+outputKey+ S3DaoImpl.extractFileNameFromPath(file));
                }else{
                    outFile = new Path("s3n://"+outputBucket+"/"+outputKey);
                }
                LOGGER.info("Out File : " + outFile.getName());
                s3OutputStream = fs.create(outFile);
                localFileInputStream = new FileInputStream(new File(file));
                IOUtils.copy(localFileInputStream, s3OutputStream);
            } catch (IOException e){
                LOGGER.error(e.getMessage());
            } finally {
                try{
                    if(s3OutputStream!=null) s3OutputStream.close();
                    if(localFileInputStream!=null) localFileInputStream.close();
                } catch (IOException e){
                    LOGGER.error(e.getMessage());
                }
            }
        }
        LOGGER.info("Copied from local to S3");
    }
}
